package com.hliushi.spark.sql

import java.io.File

import com.hliushi.spark.Utils.MyFileUtils
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.junit.Test


/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/17 21:48
 */
class ReadWrite {

  val spark: SparkSession = SparkSession.builder()
    .master("local[6]")
    .appName("reader_write")
    .getOrCreate()


  /**
   * 初体验Reader
   * #
   * SparkSession类中的 read函数原型
   * def read: DataFrameReader = new DataFrameReader(self)
   * #
   * def csv(path: String): DataFrame = csv(Seq(path): _*)
   * def csv(paths: String*): DataFrame = format("csv").load(paths : _*)
   */
  @Test
  def reader1(): Unit = {
    // 1.创建SparkSession

    /**
     * 读取数据有三种情况
     * .read.option("mode","")
     * permissive     : 当遇到损坏的记录时, 将其所有字段设置为null, 并...
     * dropMalformed  : 删除格式不正确的行
     * failFast       : 遇到格式不正确的数据时立即失败
     */

    // 2.第一种形式
    val reader = spark.read
      .format("csv") // 外部数据源的格式
      .option("mode", "FAILFAST") // 读取模式
      .option("header", value = true) // 连接外部数据源的参数, 例如JDBC的URL, 或者读取CSV文件是否引入Header
      .option("inferSchema", value = true) // inferSchema 字段类型推断
      .load("dataset/BeijingPM20100101_20151231.csv")


    reader.printSchema()

    /**
     * root
     * |-- No: integer (nullable = true)
     * |-- year: integer (nullable = true)
     * |-- month: integer (nullable = true)
     * |-- day: integer (nullable = true)
     * |-- hour: integer (nullable = true)
     * |-- season: integer (nullable = true)
     * |-- PM_Dongsi: string (nullable = true)
     * |-- PM_Dongsihuan: string (nullable = true)
     * |-- PM_Nongzhanguan: string (nullable = true)
     * |-- PM_US_Post: string (nullable = true)
     * |-- DEWP: string (nullable = true)
     * |-- HUMI: string (nullable = true)
     * |-- PRES: string (nullable = true)
     * |-- TEMP: string (nullable = true)
     * |-- cbwd: string (nullable = true)
     * |-- Iws: string (nullable = true)
     * |-- precipitation: string (nullable = true)
     * |-- Iprec: string (nullable = true)
     */

    // 3.第二种方式      这个csv("path")就是 format("csv").load("path")的封装
    spark.read
      .option("header", value = true)
      .option("inferSchema", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")
      .show(10)
  }


  /**
   * SparkSession类中的 write函数原型
   * def write: DataFrameWriter[T]
   */
  @Test
  def write1(): Unit = {
    // 解决 Could not locate executable null\bin\winutils.exe in the hadoop binaries
    System.setProperty("hadoop.home.dir", "D:\\hadoop-2.7.7\\bin\\winutils.exe")

    // 1.创建SparkSession

    // 2.读取数据集
    val df = spark.read
      .option("header", value = true)
      .option("inferSchema", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")


    //val path = "dataset/beijing_pm.json"
    val file = new File("dataset/beijing_pm.json")
    // deleteDir(file)
    // 这里把这个deleteDir方法提取到一个object单例对象中[类似Java中的静态方法], 公共使用
    MyFileUtils.deleteDir(file)


    // 3.写入数据集 --写入的格式为json
    df.write.json("dataset/beijing_pm.json")

    // 使用format设置文件格式, 使用save保存
    // mode指定了写入模式, 其中SaveMode是一个enum枚举, 定义了一些模式
    df.write.mode(SaveMode.Ignore)
      .format("json")
      .save("dataset/beijing_pm2.json")

    // 注意这个DataFrameWriter的 json方法 就是 format("json").save(path)方法的封装
    df.write.mode(SaveMode.Ignore)
      .json("dataset/beijing_pm2.json")
  }


  /**
   * Parquet是一个开源的面向列的数据存储, 它提供了多种存储优化, 允许读取单独的列非整个文件,
   * 这个不仅节省了存储空间而且提升了读取效率, 它是Spark默认的文件格式
   * Parquet为了能够保存比较复杂的数据, 并且保证性能和压缩率, 是一个不错的选择
   */
  @Test
  def parquet(): Unit = {
    // 1.读取CSV文件的数据
    val df = spark.read.option("header", value = true).csv("dataset/BeijingPM20100101_20151231.csv")
    // 2.把数据写为Parquet格式
    // df.write.format("parquet").mode(SaveMode.Overwrite).save("dataset/beijing_pm3")
    /**
     * 没有显示指定format, 只用save保存数据, 默认把数据写成Parquet格式
     * 写入模式mode, 默认是报错, [报错, 覆盖, 追加, 忽略]
     */
    df.write.mode(SaveMode.Overwrite).save("dataset/beijing_pm3")

    // 3.读取Parquet格式文件
    /**
     * 没有显示指定format, 只用load加载数据, 默认的读取格式为Parquet
     */
    spark.read.
      load("dataset/beijing_pm3").
      show()

  }


  /**
   * 表分区的概念不仅在parquet上有, 其他格式的文件也可以指定表分区
   */
  @Test
  def parquetPart(): Unit = {
    // 1.读取数据
    val dataFrame = spark.read
      .option("header", value = true)
      .option("inferSchema", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")

    // 2.写文件
    /**
     * 注意这个 partitionBy() 方法中参数字段是有顺序的
     */
    dataFrame.write
      .mode("overwrite")
      .partitionBy("year", "month")
      .save("dataset/beijing_part_pm4")


    // 3.读文件, 自动发现分区
    /**
     * 写分区表的时候, 分区列不会包含在生成的文件中
     * 直接通过文件来进行读取的话, 分区信息会丢失
     * #
     * spark sql 读取整个数据集时, 会自动发现分区
     * #
     * parquet("dataset/beijing_part_pm4/year=2010/month=1")
     * 把分区的数据集中的某一个区单独做一个整个数据集读取, 没有分区信息, 自然也不会进行分区发现
     */
    spark.read
      //.parquet("dataset/beijing_part_pm4/year=2010/month=1")
      .parquet("dataset/beijing_part_pm4")
      .printSchema()
  }

  /**
   * 读取JSON文件
   * 注意: 不支持一条数据记录跨越多行, 可以通过配置multiLine为true来进行更改
   * #
   * .    默认支持单行
   * .     {"DEPTNO": 10,"DNAME": "ACCOUNTING","LOC": "NEW YORK"}
   * .    默认不支持多行
   * .    {
   * .      "DEPTNO": 10,
   * .      "DNAME": "ACCOUNTING",
   * .      "LOC": "NEW YORK"
   * .    }
   */
  @Test
  def json(): Unit = {
    val df = spark.read
      .option("inferSchema", value = true)
      .option("header", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")

    // 2.写文件, 写成格式为json
    /**
     * 保存为JSON格式的文件有一个细节需要注意, 这个JSON格式的文件中, 每一行是一个独立的JSON
     * 但整个文件并不是一个JSON字符串, 所以这种文件格式很多时候被称为JSON Line文件, 有时候后缀名也会变为jsonl
     */
    df.write.mode(SaveMode.Overwrite).json("dataset/beijing_pm5.json")

    // 3.读json文件
    val jsonDF = spark.read.json("dataset/beijing_pm5.json")
    jsonDF.printSchema()
    jsonDF.show()
  }

  /**
   * 函数原型
   * def toJSON: Dataset[String]
   * toJSON的场景:
   * .  处理完了以后, DataFrame中如何是一个对象, 如果其他的系统只支持JSON格式的数据
   * .  SparkSQL如果和这种系统进行整合的时候, 就需要进行转换
   * #
   * 从消息队列中取出JSON格式的数据, 需要使用Spark进行处理转为RDD或者Dataset, 去读取其中的内容
   * 这个时候一条数据就是一个JSON格式的字符串
   */
  @Test
  def jsonSkill(): Unit = {
    val df = spark.read
      .option("inferSchema", value = true)
      .option("header", value = true)
      .csv("dataset/BeijingPM20100101_20151231.csv")

    // 2.
    df.toJSON.show()


    /**
     * 可以直接从RDD读取JSON的DataFrame, 把RDD[String]转为Dataset[Object]
     * 函数原型
     * def json(jsonRDD: RDD[String]): DataFrame
     */
    val jsonRDD = df.toJSON.rdd
    spark.read.json(jsonRDD).show()
  }

  /**
   * 删除一个文件夹, 及其子目录中的所有文件
   *
   * @param dir
   */
  def deleteDir(dir: File): Unit = {
    val files = dir.listFiles()
    files.foreach((f: File) => {
      if (f.isDirectory) {
        deleteDir(f)
      } else {
        f.delete()
        println("delete file " + f.getAbsolutePath)
      }
    })
    dir.delete()
    println("delete dir " + dir.getAbsolutePath)
  }

  @Test
  def wordCount(): Unit = {
    val schema = StructType(
      List(
        StructField("word", StringType)
      )
    )
    import spark.implicits._

    val source = spark.sparkContext.textFile("dataset/wordcount.txt")
    val wordDF = source.flatMap((line: String) => line.split(",")).toDF("word")
    //spark.createDataFrame(rdd1, schema)


    wordDF.select($"word")
      .groupBy($"word")
      .count()
      .sort($"count".desc)
      .show()

  }

  @Test
  def wordCount2(): Unit = {
    import spark.implicits._

    val sourceDF = spark.read.text("dataset/wordcount.txt")

    val sourceDS = sourceDF.as[String]
    val wordDF = sourceDS
      .flatMap((line: String) => line.split(","))
      .toDF("word")

    wordDF.select($"word")
      .groupBy($"word")
      .count()
      .sort($"count".desc)
      .show()

  }
}